package com.yytek.common.client.net;


import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.yytek.common.client.base.IMqClient;
import com.yytek.common.client.base.MqConfig;
import com.yytek.common.client.base.MqMessage;
import com.yytek.common.client.consts.MqTopic;
import com.yytek.common.client.net.model.*;
import com.yytek.common.client.MqClientFactory;
import com.yytek.common.client.base.MessageCallBack;



import java.util.*;
import java.util.concurrent.CompletableFuture;



public class IotMqttClient implements MessageCallBack {

    //多层通配符常量
    private static final String MULTI_LEVEL_WILDCARD = "#";
    //单层通配符常量
    private static final String SINGLE_LAYER_WILDCARD = "+";

    private static final String KEY_ID = "id";

    private IMqClient client;

    private Map<String,CompletableFuture> publishCallbackMap;
    private Map<String,List<MqttCallback>> subscribeMap;


    public IotMqttClient(){
        publishCallbackMap=new HashMap<>();
        subscribeMap=new HashMap<>();
    }

    public void init(MqConfig config){
        client = MqClientFactory.getInstance().createMQTTClient(config);
        client.setMessageCallBack(this);
        client.open();
    }


    @Override
    public void onPublish(String topic, MqMessage msg, String destination) {
        List<MqttCallback> list = subscribeMap.get(topic);
        if(destination.indexOf(MULTI_LEVEL_WILDCARD) > -1 || destination.indexOf(SINGLE_LAYER_WILDCARD) > -1){
            list = subscribeMap.get(destination);
        }
        if(list!=null&&list.size()>0){
            String productId = "";
            String deviceName = "";
            if(topic.contains(MqTopic.NetTopic.TOPIC_HEAD_SYS)){
                //处理paylod转换json
                String reply = new String(msg.getPayload());
                //获取设备和产品信息
                String[] temp = topic.split("/");
                productId = temp[2];
                deviceName = temp[3];
                JSONObject jsonObject = JSON.parseObject(reply);
                //设置设备id
                jsonObject.put("deviceName",deviceName);
                jsonObject.put("productKey",productId);
                msg.setJsonObject(jsonObject);
            }
            String finalProductId = productId;
            String finalDeviceName = deviceName;
            list.parallelStream().forEach(c->c.onMessageReceived(new IotMqttContext(this,topic, finalProductId, finalDeviceName,destination),msg));
        }

        //异步调用返回
        String reply = new String(msg.getPayload());
        JSONObject jsonObject = JSON.parseObject(reply);
        String id = jsonObject.getString(KEY_ID);
        if (Objects.isNull(id)) {
            return;
        }
        CompletableFuture iotCompletableFuture = publishCallbackMap.get(id);
        if(Objects.nonNull(iotCompletableFuture)){
            iotCompletableFuture.complete(reply);
            publishCallbackMap.remove(id);
        }

    }

    public synchronized void subscribe(String topic, MqttCallback mqttCallback) {
        List<MqttCallback> list = subscribeMap.get(topic);
        if (list == null) {
            list = new ArrayList<MqttCallback>();
            subscribeMap.put(topic, list);
            client.subscribe(topic);
        }
        if (!list.contains(mqttCallback)) {
            list.add(mqttCallback);
        }
    }

    /**
     * 属性上报订阅
     *
     */
    public void subPropertyPostSubscribe(MqttCallback callback)
    {
        subscribe(String.format(MqTopic.NetTopic.PROPERTY_POST_FMT,SINGLE_LAYER_WILDCARD,SINGLE_LAYER_WILDCARD),callback);
    }

    /**
     * 属性上报回复
     *
     */
    public void propertyPostReplyPublish(String pid, String deviceName, PropertyPostReply propertyPostReply) throws Exception {

        this.publish(String.format(MqTopic.NetTopic.PROPERTY_POST_REPLY_FMT,pid,deviceName), JSON.toJSONString(propertyPostReply));
    }

    /**
     * 属性设置
     *
     */
    public CompletableFuture<PropertySetReply> propertySetPublish(String pid, String deviceName, PropertySet propertySet) throws Exception {
        this.publish(String.format(MqTopic.NetTopic.PROPERTY_SET_FMT,pid,deviceName), JSON.toJSONString(propertySet));
        CompletableFuture<PropertySetReply> future=new CompletableFuture<PropertySetReply> ();
        publishCallbackMap.put(propertySet.getId(),future);
        //future.get(1000, TimeUnit.SECONDS);
        return future;
    }

    /**
     * 属性设置返回订阅
     *
     */
    public void propertySetReplySubscribe(MqttCallback callback) {
        subscribe(String.format(MqTopic.NetTopic.PROPERTY_SET_REPLY_FMT,SINGLE_LAYER_WILDCARD,SINGLE_LAYER_WILDCARD),callback);
    }

    /**
     * 属性获取发布
     *
     */
    public CompletableFuture<PropertyGetReply> propertyGetPublish(String pid, String deviceName, PropertyGet propertyGet) throws Exception {
        this.publish(String.format(MqTopic.NetTopic.PROPERTY_GET_FMT,pid,deviceName), JSON.toJSONString(propertyGet));
        CompletableFuture<PropertyGetReply> future = new CompletableFuture<PropertyGetReply> ();
        publishCallbackMap.put(propertyGet.getId(),future);
        //Object propertyGetReply = future.get(5, TimeUnit.SECONDS);
        return future;
    }

    /**
     * 属性获取订阅
     */
    public void propertyGetReplySubscribe(MqttCallback callback) {
        subscribe(String.format(MqTopic.NetTopic.PROPERTY_GET_REPLY_FMT,SINGLE_LAYER_WILDCARD,SINGLE_LAYER_WILDCARD),callback);
    }


    /**
     * 事件上报订阅
     *
     */
    public void eventPostScribe(MqttCallback callback)
    {
        subscribe(String.format(MqTopic.NetTopic.EVENT_POST_FMT,SINGLE_LAYER_WILDCARD,SINGLE_LAYER_WILDCARD),callback);
    }

    /**
     * 事件上报回复
     *
     */
    public void eventPostReplyPublish(String pid, String deviceName, EventPostReply eventPostReply) throws Exception {

        this.publish(String.format(MqTopic.NetTopic.EVENT_POST_REPLY_FMT,pid,deviceName), JSON.toJSONString(eventPostReply));
    }

    /**
     * 批量上报订阅
     *
     */
    public void packPostScribe(MqttCallback callback)
    {
        subscribe(String.format(MqTopic.NetTopic.SUB_PROPERTY_POST_FMT,SINGLE_LAYER_WILDCARD,SINGLE_LAYER_WILDCARD),callback);
    }

    /**
     * 批量上报回复
     *
     */
    public void packPostReplyPublish(String pid, String deviceName, PackPostReply packPostReply) throws Exception {

        this.publish(String.format(MqTopic.NetTopic.SUB_PROPERTY_POST_REPLY_FMT,pid,deviceName), JSON.toJSONString(packPostReply));
    }

    /**
     * 历史数据上报订阅
     *
     */
    public void historyPostScribe(MqttCallback callback)
    {
        subscribe(String.format(MqTopic.NetTopic.SUB_HISTORY_POST_FMT,SINGLE_LAYER_WILDCARD,SINGLE_LAYER_WILDCARD),callback);
    }

    /**
     * 历史数据上报回复
     *
     */
    public void historyReplyPublish(String pid, String deviceName, HistoryPostReply historyPostReply) throws Exception {

        this.publish(String.format(MqTopic.NetTopic.SUB_HISTORY_POST_REPLY_FMT,pid,deviceName), JSON.toJSONString(historyPostReply));
    }

    /**
     * 服务调用发布
     *
     */
    public CompletableFuture<ServiceInvokeReply> serviceInvokePublish(String pid, String deviceName,String serviceName, ServiceInvoke serviceInvoke) throws Exception {
        this.publish(String.format(MqTopic.NetTopic.SERVICE_INVOKE_FMT,pid,deviceName,serviceName), JSON.toJSONString(serviceInvoke));
        CompletableFuture<ServiceInvokeReply> future = new CompletableFuture<ServiceInvokeReply> ();
        publishCallbackMap.put(serviceInvoke.getId(),future);
        //Object propertyGetReply = future.get(5, TimeUnit.SECONDS);
        return future;
    }

    /**
     * 服务调用返回订阅
     *
     */
    public void serviceInvokeReplySubscribe(MqttCallback callback) {
        subscribe(String.format(MqTopic.NetTopic.SERVICE_INVOKE_REPLY_FMT,SINGLE_LAYER_WILDCARD,SINGLE_LAYER_WILDCARD,SINGLE_LAYER_WILDCARD),callback);
    }

    /**
     * 子设备上线订阅
     *
     */
    public void subLoginSubscribe(MqttCallback callback) {
        subscribe(String.format(MqTopic.NetTopic.SUB_LOGIN_FMT,SINGLE_LAYER_WILDCARD,SINGLE_LAYER_WILDCARD),callback);
    }

    /**
     * 子设备上线回复发布
     *
     */
    public CompletableFuture<SubLoginReply> subLoginReplyPublish(String pid, String deviceName, SubLoginReply subLoginReply) throws Exception {
        this.publish(String.format(MqTopic.NetTopic.SUB_LOGIN_REPLY_FMT,pid,deviceName), JSON.toJSONString(subLoginReply));
        CompletableFuture<SubLoginReply> future = new CompletableFuture<SubLoginReply> ();
        publishCallbackMap.put(subLoginReply.getId(),future);
        //Object propertyGetReply = future.get(5, TimeUnit.SECONDS);
        return future;
    }

    /**
     * 子设备离线订阅
     *
     */
    public void subLogoutSubscribe(MqttCallback callback) {
        subscribe(String.format(MqTopic.NetTopic.SUB_LOGOUT_FMT,SINGLE_LAYER_WILDCARD,SINGLE_LAYER_WILDCARD),callback);
    }

    /**
     * 子设备离线回复发布
     *
     */
    public CompletableFuture<SubLogoutReply> subLogoutReplySubscribe(String pid, String deviceName, SubLogoutReply subLogoutReply) throws Exception {
        this.publish(String.format(MqTopic.NetTopic.SUB_LOGOUT_REPLY_FMT,pid,deviceName), JSON.toJSONString(subLogoutReply));
        CompletableFuture<SubLogoutReply> future = new CompletableFuture<SubLogoutReply> ();
        publishCallbackMap.put(subLogoutReply.getId(),future);
        //Object propertyGetReply = future.get(5, TimeUnit.SECONDS);
        return future;
    }


    public void replyPropertyPost(String productId,String deviceName, PropertyPostReply reply){
        publish(String.format(MqTopic.NetTopic.PROPERTY_POST_FMT,productId,deviceName), JSON.toJSONString(reply));
    }

    private void reply(String topic,String content){

    }

    public void publish(String topic,String content){
        client.publish(topic, content);
    }



}
